package com.weic.flink.datastream.transformation

import org.apache.flink.streaming.api.functions.ProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector

/**
 * @Auther:BigData-weic
 * @ClassName:_01SplitAndSelectOps
 * @Date:2020/12/16 19:23
 * @功能描述: split and select 拆分流
 * @Version:1.0
 */
object _01SplitAndSelectOps {
	def main(args: Array[String]): Unit = {
		val env = StreamExecutionEnvironment.getExecutionEnvironment
		val lines = env.socketTextStream("node-1",9999)
		//025|lining|sports
		val goods = lines.map(line => {
			val fields = line.split("\\|")
			val id = fields(0)
			val brand = fields(1)
			val category = fields(2)
			Goods(id, brand, category)
		})
		//定义标签
		val sportsTag = OutputTag[Goods]("sports")
		val clothingTag = OutputTag[Goods]("clothing")
		val mobileTag = OutputTag[Goods]("mobile")
		//使用process函数打标签
		val ret = goods.process(new ProcessFunction[Goods, Goods] {
			override def processElement(i: Goods, context: ProcessFunction[Goods, Goods]#Context,
										collector: Collector[Goods]): Unit = {
				i.category match {
					//为每一条数据打标签
					//模式匹配
					case "sports" => {
						context.output(sportsTag, i)
					}
					case "clothing" => {
						context.output(clothingTag, i)
					}
					case "mobile" => {
						context.output(mobileTag, i)
					}
					case _ => {
						collector.collect(i)
					}
				}
			}
		})
		//选择标签
		ret.getSideOutput(sportsTag).print("sports:::")
		ret.getSideOutput(mobileTag).print("mobile:::")
		//execute执行
		env.execute(s"${_01SplitAndSelectOps.getClass.getSimpleName}")
	}

}

case class Goods(id: String, brand: String, category: String)
